package com.dimld.connect.websocket;

import com.dimld.connect.*;
import io.undertow.websockets.core.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class WebSocketClientHandler extends AbstractReceiveListener implements MsgReaderI, ConnectorI, Serializable {
    private static final Logger log = LoggerFactory.getLogger("ws");

    private transient Thread     messageSenderThread;
    private transient Thread     messageReaderThread;
    private transient Timer      timeoutTimer;
    private transient MsgReaderI msgReader;

    private           String                                bindServiceUuid;
    private           Object                                data;
    private           String                                uuid;
    private           boolean                               state;
    private final     long                                  createTime;
    private final     AtomicInteger                         stale       = new AtomicInteger(0);
    private           WebSocketChannel                      channel;
    private final     WebSocketClientHandler                webSocketClientHandler;
    private           LinkedBlockingQueue<MessageStructure> msgQueue    = new LinkedBlockingQueue<>();
    private transient BlockingQueue<String>                 readerQueue = new LinkedBlockingQueue<>();

    public WebSocketClientHandler(WebSocketChannel webSocketChannel, String uuid, MsgReaderI msgReader, final int timeout, String serviceUUID) {
        this.channel = webSocketChannel;
        webSocketClientHandler = this;
        this.uuid = uuid;
        this.msgReader = msgReader;
        this.bindServiceUuid = serviceUUID;
        createTime = System.currentTimeMillis();
        setStateTrue();
        messageSenderThread = new Thread(new MessageSenderThread());
        messageSenderThread.start();
        messageReaderThread = new Thread(new MessageReaderThread(this));
        messageReaderThread.start();
        timeoutTimer = new Timer();
        timeoutTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                int staleTime = stale.incrementAndGet();
                if(staleTime >= timeout) {
                    log.info("[WEBSOCKET][]客户端长时间未操作，服务器强制断开连接");
                    close();
                    timeoutTimer.cancel();
                }
            }
        }, 1000, 1000);
        ConnectContainer.put(this);
        if(msgReader != null) {
            msgReader.connect(this);
        } else {
            log.warn("[WEBSOCKET][" + getFullIP() + "]没有实现开启连接监听");
        }
    }

    @Override
    protected void onFullTextMessage(WebSocketChannel channel, BufferedTextMessage bufferedTextMessage) {
        this.stale.set(0);
        String message = bufferedTextMessage.getData();
        readerQueue.add(message);
    }

    @Override
    protected void onCloseMessage(CloseMessage cm, WebSocketChannel channel) {
        log.debug("关闭连接code - {} reason - {} channel - {}", cm.getCode(), cm.getReason(), channel.getSourceAddress());
        close();
    }

    @Override
    public void send(String receiveMsg) {
        try {
            if(ConnectContainer.get(this.uuid) != null) {
            }
        } catch(Exception e) {
            log.error("线程被中断，发送消息失败");
        }
    }

    @Override
    public void close() {
        try {
            if(ConnectContainer.get(this.uuid) != null) {
                log.debug("[WEBSOCKET][" + getFullIP() + "]添加关闭连接消息到队列");
            }
        } catch(Exception e) {
            log.error("异常", e);
        }
    }

    @Override
    public String getConnType() {
        return ConnectType.WEBSOCKET;
    }

    @Override
    public String getUuid() {
        return uuid;
    }

    public void setUuid(String uuid) {
        this.uuid = uuid;
    }

    @Override
    public MsgReaderI getReader() {
        return msgReader;
    }

    @Override
    public boolean isConnected() {
        return state;
    }

    @Override
    public String getFullIP() {
        if(channel != null) {
            return channel.getSourceAddress().toString();
        } else {
            return "";
        }
    }

    @Override
    public long getCreateTimestamp() {
        return createTime;
    }

    @Override
    public Object getData() {
        return data;
    }

    @Override
    public void setData(Object data) {
        this.data = data;
    }

    public void setStateFalse() {
        state = false;
    }

    public void setStateTrue() {
        state = true;
    }

    public AtomicInteger getStale() {
        return stale;
    }

    public void setStale(int stale) {
        this.stale.set(stale);
    }

    private void stop() {
        timeoutTimer.cancel();
        if(messageReaderThread != null && messageReaderThread.isAlive()) {
            messageReaderThread.interrupt();
        }
        try {
            if(channel != null) {
                log.debug("[{}] channel closing...", this.uuid);
                channel.sendClose();
            } else {
                log.debug("[{}] channel is null", this.uuid);
            }
        } catch(IOException e) {
            log.error("IO异常，连接已关闭。{}", e.getMessage());
        } catch(Exception e) {
            log.error("出现异常", e);
        }
        destroy();
        setStateFalse();
        ConnectContainer.remove(getUuid());
    }

    public void destroy() {
    }

    @Override
    public String getBindServiceUuid() {
        return bindServiceUuid;
    }

    @Override
    public void connect(ConnectorI paramConnectorI) {

    }

    @Override
    public void read(String paramString, ConnectorI paramConnectorI) {
    }

    @Override
    public void disconnect(ConnectorI paramConnectorI) {
    }

    class MessageReaderThread implements Runnable {
        WebSocketClientHandler webSocketClient;

        MessageReaderThread(WebSocketClientHandler webSocketClient) {
            this.webSocketClient = webSocketClient;
        }

        @Override
        public void run() {
            Thread.currentThread().setName("MessageReader_" + getUuid());
            log.debug("[WEBSOCKET][" + getFullIP() + "]读取消息队列启动");
            while(true) {
                try {
                    String message = readerQueue.take();
                    if(msgReader != null) {
                        msgReader.read(message, webSocketClient);
                    }
                } catch(InterruptedException e) {
                    log.error("被中断，退出线程");
                    break;
                } catch(Exception e) {
                    log.error("其他异常", e);
                    break;
                }
            }
            log.debug("[WEBSOCKET][{}}]读取消息队列线程停止", getFullIP());
        }
    }

    class MessageSenderThread implements Runnable {

        @Override
        public void run() {
            Thread.currentThread().setName("MessageThread_" + getUuid());
            log.info("[WEBSOCKET][" + getFullIP() + "]发送消息线程启动");
            while(true) {
                try {
                    MessageStructure webSocketMessageStructure = msgQueue.take();
                    if(webSocketMessageStructure.getType() == 1) {
                        log.info("[WEBSOCKET][" + getFullIP() + "]发送消息：" + webSocketMessageStructure.getMessage());
                        WebSockets.sendText(webSocketMessageStructure.getMessage(), channel, null);
                    } else {
                        stop();
                        if(msgReader != null)
                            msgReader.disconnect(webSocketClientHandler);
                        break;
                    }
                } catch(Exception e) {
                    log.info("[WEBSOCKET][" + getFullIP() + "]消息发送终止，线程被中断");
                    break;
                }
            }
            log.info("[WEBSOCKET][" + getFullIP() + "]发送消息线程停止");
        }
    }
}
